package de.fhkoeln.avs.parallelbreadthfirstsearch.infinitegraph;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.infinitegraph.ConfigurationException;
import com.infinitegraph.StorageException;
import de.fhkoeln.avs.graphbuild.infinitegraph.storage.AvsVertex;


public class LocalThreadCrawler extends Thread implements Crawler{

	private Map<String, AvsVertex> sendBuf; //Name des Knoten als Key
//	private Map<String, AvsVertex> recvBuf; //Name des Knoten als Key
	private Map<String, List<AvsVertex>> tBufs; //Idetifier des Crawlers als key
	private Map<String, Crawler> crawlers;
		
	private String identifier;
	private AtomicInteger receivedTBufs;
	
	private PFBSCallback callbackToMainThread;
	private List<AvsVertex> toHandlerVertices;
	
	private ExecutorService executor;	
	
	//Technical
	private static Logger logger = LoggerFactory.getLogger(LocalThreadCrawler.class);
	private boolean stopped = true;
	
	public LocalThreadCrawler(String identifier){
		sendBuf = new ConcurrentHashMap<String, AvsVertex>();
//		recvBuf = new HashMap<String, AvsVertex>();
		tBufs = new ConcurrentHashMap<String, List<AvsVertex>>();
		this.identifier = identifier;
		receivedTBufs = new AtomicInteger(0); 
		crawlers = new HashMap<String, Crawler>();
		toHandlerVertices = new ArrayList<AvsVertex>();
	}

	@Override
	public void setCrawlers(List<Crawler> crawlers){
		for (Crawler bro : crawlers) {
			tBufs.put(bro.getIdentifier(), Collections.synchronizedList(new LinkedList<AvsVertex>()));
			this.crawlers.put(bro.getIdentifier(), bro);
		}
		executor = Executors.newFixedThreadPool(this.crawlers.size());
	}
	
	@Override
	public String getIdentifier() {
		return this.identifier;
	}

	
	@Override
	public void run(){
		while(!stopped);
//		logger.info("Crawler with id: "+ getIdentifier()+"successfully to stop.");
	}
	
	@Override
	public void startCrawler() {
		stopped = false;
		this.start();
	}

	@Override
	public void stopCrawler() {
		stopped = true;	
//		logger.info("Crawler with id: "+ getIdentifier()+"attempted to stop.");
	}

	@Override
	public  void distributeChildVertices(AvsVertex vertex) throws Exception{
		
//		GraphDatabase database = DatabaseConnection.getConnection();		
//		Transaction tx = database.beginTransaction();
//		AvsVertex vertex = (AvsVertex) database.getNamedVertex(vertexName);//it.next();
		for (AvsVertex child : vertex.getChilds()) {
			String owner = child.getOwner();
			tBufs.get(owner).add(child);
		}
//		tx.commit();
		callbackToMainThread.buildTBufsFinished();
	}

	@Override
	public  void receiveTBuffer(List<AvsVertex> tBuf, Integer crawlerInUse, CrawlerCallback crawlerCallback) throws Exception{	
		mergeSendBufWithTBuf(tBuf);
		Integer value = receivedTBufs.incrementAndGet();
		if(value == (crawlerInUse)){
			receivedTBufs.set(0);
			List<AvsVertex> ownTBuf = tBufs.get(this.getIdentifier());
			mergeSendBufWithTBuf(ownTBuf);
		}
		crawlerCallback.mergedTBufFinished();
	}
	
	private  void asyncCallReceiveTBuffer(final Semaphore semaphore,
			final List<AvsVertex> tBuf, final Crawler crawler, Integer crawlerInUse) throws Exception {
		
		crawler.receiveTBuffer(tBuf, crawlerInUse, createCrawlerCallback(semaphore, crawler));
	}

	private  CrawlerCallback createCrawlerCallback(final Semaphore semaphore,
			final Crawler crawler) {
		return new CrawlerCallback() {

			@Override
			public void mergedTBufFinished() {
				semaphore.release();
			}
		};
	}

	private  void mergeSendBufWithTBuf(List<AvsVertex> tBuf) throws StorageException, ConfigurationException {
		for (AvsVertex avsVertex : tBuf) {
			sendBuf.put(avsVertex.getName(), avsVertex);
		}
	}

	@Override
	public void cleanCrawler() {
		sendBuf.clear();
		toHandlerVertices.clear();
		for(List<AvsVertex> tBuf : tBufs.values()){
			tBuf.clear();
		}
	}

	@Override
	public  void sendTBufs(final Integer crawlerInUse) throws InterruptedException {
		final Semaphore semaphore = new Semaphore(crawlerInUse);
		for (String crawlerId : tBufs.keySet()) {
			semaphore.acquire();
			final List<AvsVertex> tBuf = tBufs.get(crawlerId);
			final Crawler crawler = crawlers.get(crawlerId);
			executor.submit(new Runnable() {
				@Override
				public void run() {
					try {
						asyncCallReceiveTBuffer(semaphore, tBuf,crawler, crawlerInUse);
					} catch (Exception e) {
						e.printStackTrace();
						throw new RuntimeException("Algorithmus aufgrund eines Fehlers abgebrochen");
					}
				}
			});	
		}
		semaphore.acquireUninterruptibly(crawlerInUse);
		callbackToMainThread.sendTBufsFinished(sendBuf);
	}

	@Override
	public void setCallback(PFBSCallback callback) {
		this.callbackToMainThread = callback;
	}
}
